/*
 * Copyright 2011-2012 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.sabayframework.channel.jdbc;

import java.sql.Connection;
import java.util.LinkedList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

import org.sabayframework.channel.Channel;
import org.sabayframework.channel.Lock;

/**
 * Channel for asynchronous query execution using JDBC connection.
 * Channel uses an instance of ExecutorService for executing queries in blocking threads.
 * The ExecutorService can be shared between multiple channels but Connection should be used 
 * by a single JDBCChannel instance. 
 * @author Alex Antonau
 */
public class JDBCChannel implements Channel<JDBCQuery> {

	private final ExecutorService executor;
	private final Connection connection;
	private final LinkedList<JDBCQuery> queue;
	private final ReentrantLock lock;
	private final Condition notEmpty;	
    private volatile boolean closed;
    private volatile Throwable failure;
	
	public JDBCChannel(ExecutorService executor, Connection connection) {
		super();
		this.executor = executor;
		this.connection = connection;
		this.lock = new ReentrantLock();
		this.notEmpty = lock.newCondition();
		this.queue = new LinkedList<JDBCQuery>();
		
	}
	private final class ReadLock implements Lock<JDBCQuery>
	{
		private JDBCQuery[] locked;
		ReadLock(final JDBCQuery[] locked) {
			this.locked = locked; 
		}

		@Override
		public int size() {
			return locked == null ? -1: locked.length;
		}

		@Override
		public JDBCQuery get(int index) {
			return locked[index];
		}

		@Override
		public void set(int index, JDBCQuery elem) {
		}

		@Override
		public JDBCQuery[] toArray() {
			return locked;
		}

		@Override
		public void unlock() {
			locked = null;
			lock.unlock();
		}
		@Override
		public void cancel() {
			for(JDBCQuery c:locked)queue.add(c);
			locked = null;
			lock.unlock();
		}
	}


	private final class WriteLock implements Lock<JDBCQuery>{
		private JDBCQuery[] locked;
		WriteLock(JDBCQuery[] locked) {
			this.locked = locked;
		}

		@Override
		public int size() {
			return locked == null ? -1: locked.length;
		}

		@Override
		public JDBCQuery get(int index) {
			return locked[index];
		}

		@Override
		public void set(int index, JDBCQuery elem) {
			locked[index] = elem;
		}

		@Override
		public JDBCQuery[] toArray() {
			return locked;
		}

		@Override
		public void unlock() {
			if(locked !=null){
				for(JDBCQuery q : locked) processQuery(q);
				notEmpty.signalAll();
				lock.unlock();
				locked = null;
			}
		}
		@Override
		public void cancel() {
			if(locked !=null){
				locked = null;
				lock.unlock();
			}
		}
		
	}
	private void processQuery(JDBCQuery c){
		c.setConnection(connection);
		c.setFuture(executor.submit(c));
		queue.add(c);
	}
	
	@Override
	public Lock<JDBCQuery> getReadLock(final int size) {
		if(size < 0) throw new IllegalArgumentException(String.format("Negative lock size %d", size));
		ReadLock result = null;
		if(lock.tryLock()){
			try{
				int available = 0;
				for(int i=0;available < size && i<queue.size();++i){
					JDBCQuery c = queue.get(i);
					final Future<?> future = c.getFuture();
					if(future.isDone()){
						++available;
					}
				}
				if(available >= size){
					int lockedIndex = 0;
					final JDBCQuery[] locked = new JDBCQuery[size];
					for(int i=0;lockedIndex < size && i<queue.size();++i){
						JDBCQuery c = queue.get(i);
						final Future<?> future = c.getFuture();
						if(future.isDone()){
							locked[lockedIndex++] = c;
							queue.remove(i);
							--i;
						}
					}
					result = new ReadLock(locked);
				}
			}finally{
				if(result == null)lock.unlock();
			}
		}
		return result;
	}

	private int firstDoneIndex(){
		for(int i=0;i<queue.size();++i){
			JDBCQuery c = queue.get(i);
			final Future<?> future = c.getFuture();
			if(future.isDone()){
				return i;
			}
		}
		return -1;
	}
	@Override
	public JDBCQuery remove() throws IllegalStateException {
		if(lock.tryLock()){
			try{
				int index = firstDoneIndex();
				if(index!=-1) return queue.remove(index);
			}finally{
				lock.unlock();
			}
		}
		throw new IllegalStateException();
	}

	@Override
	public Lock<JDBCQuery> getWriteLock(final int size) {
		if(size < 0) throw new IllegalArgumentException(String.format("Negative lock size %d", size));
		WriteLock result = null;
		if(lock.tryLock()){
			try{
				if(!closed){
					result = new WriteLock(new JDBCQuery[size]);
				}
			} finally {
				if(result == null) lock.unlock();
			}
		}
		return result;
	}
	
	@Override
	public void add(JDBCQuery elem) throws IllegalStateException {
		if(!closed && lock.tryLock()){
			try{
				processQuery(elem);
			}finally{
				lock.unlock();
			}
		}else throw new IllegalStateException();
	}

	@Override
	public boolean isEmpty() {
		return size() == 0;
	}

	@Override
	public boolean isFull() {
		return false;
	}

	@Override
	public int size() {
        lock.lock();
        try {
        	return queue.size();
        } finally {
            lock.unlock();
        }
	}

	@Override
	public int capacity() {
		return Integer.MAX_VALUE;
	}

	@Override
	public int freeSize() {
		return Integer.MAX_VALUE;
	}

	@Override
	public void close() {
		if(closed) return;
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
       		closed = true;
       		notEmpty.signalAll();
        } finally {
            lock.unlock();
        }
	}

	@Override
	public boolean isClosed() {
		return closed;
	}

	@Override
	public void fail(Throwable cause) {
		this.failure = cause;
		close();
	}

	@Override
	public boolean hasFailed() {
		return failure!=null;
	}

	@Override
	public Throwable getFailureCause() {
		return failure;
	}

	@Override
	public Lock<JDBCQuery> waitForReadLock(final int size, long timeout)	throws InterruptedException {
		final long startTime = System.currentTimeMillis();
		long remaining = timeout;
		while(remaining>0 && !closed && lock.tryLock(remaining, TimeUnit.MILLISECONDS)){
			Future<?> future = null;
			try{
		        final Lock<JDBCQuery> readLock = getReadLock(size);
				if(readLock!=null) return readLock;
				remaining = timeout - (System.currentTimeMillis() - startTime);
				if(queue.size() > 0) future = queue.getFirst().getFuture();
				else
				{
					notEmpty.await(remaining, TimeUnit.MILLISECONDS);
				}
			} finally {
            	lock.unlock();
            }
			if(future!=null && remaining>0){
				try {
					future.get(remaining, TimeUnit.MILLISECONDS);
				} catch (Exception e) {}
			} 
			remaining = timeout - (System.currentTimeMillis() - startTime);
		}
		return getReadLock(size);
	}

}
